/**
 * Copyright 2009 Takahiko Ito
 * 
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0 
 *        
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License.
 */
package org.unigram.oluolu.rqe.selector;

import org.apache.hadoop.conf.Configuration;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

/**
 * Select the pair of query is related or not by the value of 
 * character based cosine value between two query strings.
 */
public class CosineSelector implements ISelector {

    /** threshold. */
    private float releatednessThreshold;

    /**
     * Constructor.
     *
     * @param jc Configuration
     */
    public CosineSelector(final Configuration jc) {
        releatednessThreshold = jc.getFloat(
                RQE_COSINE_THRESHOLD,
                DEFAULT_RQE_COSINE_THRESHOLD);
    }

    /**
     * Validate the pair queries are related or not.
     *
     * @param strOne query string
     * @param strTwo query string
     * @return true when the queries are related otherwise return false
     */
    public final boolean run(final String strOne, final String strTwo) {
        
        /* length of the queries should be almost the same */        
        if (Math.abs(strOne.length() - strTwo.length()) > 3) {
            return false;
        }
        
        /* split input strings into arrays */
        Pattern pattern = Pattern.compile("");
        String[] strOneArray = pattern.split(strOne);
        String[] strTwoArray = pattern.split(strTwo);

        /* convert array into map */
        Map vectorOne = createVector(strOneArray);
        Map vectorTwo = createVector(strTwoArray);

        /* compute the inner products and norms */
        double ip = (double) calcInnerProduct(vectorOne, vectorTwo);
        double normOne = (double) calcNorm(vectorOne);
        double normTwo = (double) calcNorm(vectorTwo);
        double norm = (normOne * normTwo);

        if ((ip / norm) > releatednessThreshold) { 
            return true;
        } 
        return false;
        
    }

    /**
     * Calculate inner product between two vectors.
     *
     * @param vectorOne vector extracted first query string
     * @param vectorTwo vector extracted second query string
     * @return inner product between vectorOne and vectorTwo
     */
    private long calcInnerProduct(final Map vectorOne, final Map vectorTwo) {
        long ip = 0;
        Set demensions = vectorOne.keySet();
        Iterator iterator = demensions.iterator();
        while (iterator.hasNext()) {
            String demension = (String) iterator.next();
            if (vectorTwo.containsKey(demension)) {
                Long vOne = (Long) vectorOne.get(demension);
                Long vTwo = (Long) vectorTwo.get(demension);
                ip += (vOne * vTwo);
            }
        }
        return ip;
    }

    /**
     * Calculate norm for vector.
     *
     * @param vector character based query string vector
     * @return norm
     */
    private double calcNorm(final Map vector) {
        long norm = 0;
        Set demensions = vector.keySet();
        Iterator iterator = demensions.iterator();

        while (iterator.hasNext()) {
            String demension = (String) iterator.next();
            norm += (Long) vector.get(demension);
        }
        return Math.sqrt(norm);
    }

    /**
     * Create vector for assigned query string. 
     *
     * @param array query string consists of the elements of array
     * @return vector representation of query string
     */
    private Map createVector(final String[] array) {
        Map<String, Long> rtMap = new HashMap<String, Long>();

        for (int i = 0; i < array.length; i++) {
            String demension = array[i];
            if (rtMap.containsKey(demension)) {
                Long count = (Long) rtMap.get(demension);
                rtMap.put(demension, count);
                count += 1;
            } else {
                rtMap.put(new String(demension), new Long(1));
            }
        }
        return rtMap;
    }
    
    /** symbol: cosine threshold. */
    public static final String RQE_COSINE_THRESHOLD
        = "oluolu.rqe.cosine.threshold";
    
    /** default: threshold of cosine selector. */
    public static final float DEFAULT_RQE_COSINE_THRESHOLD = 0.8F;    
            
}
